package com.yz.kafka.ks;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

/**
 * @author yuanzheng
 * @since 2021/4/5
 */
public class LogProcessor implements Processor<byte[], byte[]> {
    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
    }

    @Override
    public void process(byte[] key, byte[] value) {
        String input = new String(value);
        // 如果包含“>>>”则只保留该标记后面的内容
        String repStr = ">>>";
        if (input.contains(repStr)) {
            input = input.split(repStr)[1].trim();
        }
        // 输出到下一个topic
        context.forward("logProcessor".getBytes(), input.getBytes());
    }

    @Override
    public void punctuate(long l) {

    }

    @Override
    public void close() {

    }
}
